package com.yn.thread;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * @ClassName: MQThread
 * @Description:
 * @author: 王胜
 * @date 2021/6/30
 * @Version: 1.0
 **/
public class MQCustomerThread  extends Thread{
    private String queueName="";
    private String host = "127.0.0.1";
    private int port = 5672;
    private String username = "guest";
    private String pwd = "guest";

    public MQCustomerThread(){
    }
    public MQCustomerThread(String queueName){
        this.queueName=queueName;
    }

    public MQCustomerThread(String queueName,String host,int port,String username,String pwd){
        this.queueName=queueName;
        this.host=host;
        this.port=port;
        this.username=username;
        this.pwd=pwd;
    }
    @Override
    public void run(){
        //amqp协议基于tcp/ip协议
        //ip,port
        //创建连接工厂
        ConnectionFactory factory=new ConnectionFactory();
        factory.setHost(host);
        factory.setPort(port);
        factory.setUsername(username);
        factory.setPassword(pwd);
        Connection connection=null;
        Channel channel =null;
//        获取连接
        try {
            connection= factory.newConnection("消费者");
            //通过连接获取通道
            channel=connection.createChannel();
            String queue1=queueName;
            //发送消息给队列
            channel.basicConsume(queue1, true,
                    (consumerTag, delivery) ->{
                        System.out.println(queueName+" 收到一条消息：" + new String(delivery.getBody(),"UTF-8")) ;
                        },
                    consumerTag -> System.out.println(queueName+" 出错取消"));
            System.out.println(queueName+" receive successfully");
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }finally {
            //关闭连接
            if(connection!=null&&connection.isOpen()){
                try {
                    connection.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            //关闭通道
            if(channel!=null&&channel.isOpen()){
                try {
                    channel.close();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}